package com.bridgeintelligent.tag.bulkload.service.dimfree;

import com.bridgeintelligent.tag.bulkload.service.looklike.HdfsService;
import com.bridgeintelligent.tag.constants.PublicConstant;
import com.bridgeintelligent.tag.constants.bulkload.FileToHdfsMsg;
import com.bridgeintelligent.tag.constants.bulkload.MQConfig;
import com.bridgeintelligent.tag.mapper.IssueTaskMapper;
import com.bridgeintelligent.tag.mapper.UploadHfdsLogMapper;
import com.bridgeintelligent.tag.utils.DateHelper;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.stereotype.Component;

/**
 * @Author：liwei
 * @Date：2023/11/13 14:52
 * @Desc： 文件备份Hdfs 这里的msg中的OriginalFile的值为文件全路径
 *         type:1定制标签
 */
@Slf4j
@Component
@RocketMQMessageListener(topic = MQConfig.FILE_BAKUP_HDFS_TOPIC, consumerGroup = MQConfig.FILE_BAKUP_HDFS_GROUP,consumeThreadMax=1)
public class FileBakupListener implements RocketMQListener<String> {
    private ObjectMapper objectMapper;
    private HdfsService hdfsService;
    private UploadHfdsLogMapper mapper;
    private IssueTaskMapper issueTaskMapper;



    public FileBakupListener(ObjectMapper objectMapper, HdfsService hdfsService,UploadHfdsLogMapper mapper,IssueTaskMapper issueTaskMapper) {
        this.objectMapper = objectMapper;
        this.hdfsService = hdfsService;
        this.mapper = mapper;
        this.issueTaskMapper = issueTaskMapper;
    }

    @Override
    public void onMessage(String s) {
        FileToHdfsMsg msg = null;
        try {
            msg = objectMapper.readValue(s, FileToHdfsMsg.class);
            log.info("=====>备份文件写入Hdfs接收消息：FILE_BAKUP_HDFS_TOPIC:{}>",msg);
            Thread.sleep(1000);
            mapper.del(msg.getFilePath());

            //幂等性控制，防止单次消费处理时间大于MQ默认15分钟触发重试机制
            int count = issueTaskMapper.transNum(msg.getUniqueIdentify());
            log.info("======>【静态客群ToHdfs】当前线程：{}，处理：{}，获取count:{}",Thread.currentThread().getName(),msg.getUniqueIdentify(),count);
            if (count > 0){
                log.error("任务：{}已经处理过",msg.getUniqueIdentify());
                return;
            }
            issueTaskMapper.addTransNum(msg.getUniqueIdentify());

            hdfsService.uploadHdfs(msg.getFilePath(),msg.getType());
        } catch (JsonProcessingException e) {
            log.error("=====>备份文件写入Hdfs消息[{}]转换失败！！！",msg);
        } catch (Exception e){
            //失败入库记录
            mapper.add(msg.getFilePath(), String.valueOf(msg.getType()),PublicConstant.TWO,DateHelper.currentDateTime());
            log.error("=====>备份文件[{}]写HDFS文件操作失败！！！",msg);
            e.printStackTrace();
        }

    }

}
